#include <brpc/server.h>
#include <butil/logging.h>
#include <map>
//实现语音识别子服务
#include "mysql_chat_session_member.hpp"
#include <boost/date_time/posix_time/posix_time.hpp>
#include "rabbitmq.hpp"
#include "utils.hpp"
#include "etcd.hpp"     //服务注册
#include "logger.hpp"   //日志模块
#include "channel.hpp"

#include "base.pb.h"
#include "transmite.pb.h"
#include "user.pb.h"


namespace xchat{
//proto rpc TransmiteService ; package xchat
class TransmiteServiceImpl : public xchat::MsgTransmitService
{
public:
    TransmiteServiceImpl(const std::string &user_service_name, const ServiceManager::ptr &mm_channels,
                        const std::shared_ptr<odb::core::database> &mysql_client, 
                        const std::string exchange_name, const std::string &binding_key,
                        const MQClient::ptr &mq_client)
            :_user_service_name(user_service_name)
            ,_mm_channels(mm_channels)
            ,_mysql_session_member_table(std::make_shared<ChatSessionMemberTable>(mysql_client))
            ,_exchange_name(exchange_name)
            ,_binding_key(binding_key)
            ,_mq_client(mq_client)
    {}
    ~TransmiteServiceImpl(){}
    //组装一个完整的消息
    void GetTransmitTarget(google::protobuf::RpcController* controller,
                        const ::xchat::NewMessageReq* request,
                        ::xchat::GetTransmitTargetRsp* response,
                        ::google::protobuf::Closure* done)
    {
        auto error_response = [this, response](const std::string &rid, const std::string &error_msg) -> void
        {
            response->set_request_id(rid);
            response->set_success(false);
            response->set_errmsg(error_msg);
        };   
        brpc::ClosureGuard rpc_guard(done);
        // 从请求中 获取关键信息
        std::string rid = request->request_id();
        std::string uid = request->user_id();
        std::string chat_ssid = request->chat_session_id();
        const MessageContent &content = request->message();
        // 进行消息组织 消息id 发送者 所属会话 产生时间 消息内容
        auto channel = _mm_channels->choose(_user_service_name);
        if(!channel)
        {
            LOG_ERROR("{}-{}没有可供访问的用户子服务节点", rid, _user_service_name);
            return error_response(rid, "没有可供访问的用户子服务节点");
        }
        UserService_Stub stub(channel.get());
        GetUserInfoReq get_uinfo_req;

        get_uinfo_req.set_request_id(rid);
        get_uinfo_req.set_user_id(uid);

        std::unique_ptr<xchat::GetUserInfoRsp> get_uinfo_rsp(new xchat::GetUserInfoRsp());
        std::unique_ptr<brpc::Controller> cntl(new brpc::Controller());
        stub.GetUserInfo(cntl.get(), &get_uinfo_req, get_uinfo_rsp.get(), nullptr); // 同步调用
        if(cntl->Failed())
        {
            LOG_ERROR("用户子服务调用失败: {} 请求id: {} 错误原因: {}", uid, get_uinfo_req.request_id(), cntl->ErrorText());
            return error_response(get_uinfo_req.request_id(), "用户子服务调用失败");
        }
        if(!get_uinfo_rsp->success())
        {
            LOG_ERROR("用户子服务调用失败: {} 请求id: {} 错误原因: {}", uid, get_uinfo_req.request_id(), get_uinfo_rsp->errmsg());
            return error_response(get_uinfo_req.request_id(), "用户子服务调用失败");
        }
        MessageInfo message;
        message.set_message_id(uuid());
        message.set_chat_session_id(chat_ssid);
        message.set_timestamp(time(nullptr));  //时间戳 +8小时
        message.mutable_sender()->CopyFrom(get_uinfo_rsp->user_info());
        message.mutable_message()->CopyFrom(content);

        // 并且获取消息转发客户端用户列表
        auto target_list = _mysql_session_member_table->members(chat_ssid);
        
        // 将封装完毕的消息 发布到消息队列 待消息存储子服务持久化
        bool ret = _mq_client->publish(_exchange_name, message.SerializeAsString(), _binding_key);//序列化为字符串
        if(!ret)
        {
            LOG_ERROR("消息队列客户端发布消息失败: {} 请求id: {}", rid, message.message_id());
            return error_response(rid, "消息队列客户端发布消息失败");
        }

        // 组装响应
        response->set_request_id(rid);
        response->set_success(true);
        response->mutable_message()->CopyFrom(message); 
        for(auto &id : target_list)
        {
            response->add_target_id_list(id);
        }
    }
private:
    //用户子服务调用相关信息
    std::string _user_service_name;
    ServiceManager::ptr _mm_channels; //信道管理对象
    //聊天会话成员表操作句柄
    ChatSessionMemberTable::ptr _mysql_session_member_table; //用户表
    //消息队列客户端 消息句柄
    std::string _exchange_name;
    std::string _binding_key;
    MQClient::ptr _mq_client; 
};
class TransmiteServer
{
    public:
        using ptr = std::shared_ptr<TransmiteServer>;
        TransmiteServer(const std::shared_ptr<odb::core::database> &mysql_client, const Discovery::ptr &service_discovery, 
            const Registry::ptr &reg_client, const std::shared_ptr<brpc::Server> &server)
        :_service_discovery(service_discovery)
        ,_reg_client(reg_client)
        ,_mysql_client(mysql_client)
        ,_rpc_server(server)
        {}
        ~TransmiteServer()
        {}

        //搭建rpc服务器 并启动服务器
        void start()
        {
            _rpc_server->RunUntilAskedToQuit();
        }
    private:
        Discovery::ptr _service_discovery; //服务发现对象
        Registry::ptr _reg_client;  //注册中心
        std::shared_ptr<odb::core::database> _mysql_client; //数据库连接
        std::shared_ptr<brpc::Server> _rpc_server;       //rpc服务
};
//建筑师模式
class TransmiteServerBuilder
{
    public:
        //用于构造服务发现客户端对象 & 信道管理对象
        void make_discovery_object(const std::string &reg_host, const std::string &base_service_name
                                    , const std::string &user_service_name)
        {
            _user_service_name = user_service_name;
            _mm_channels = std::make_shared<ServiceManager>();
            _mm_channels->declared(user_service_name);   //声明关心那些服务
            auto put_cb = std::bind(&ServiceManager::onServiceOnline, _mm_channels.get(), std::placeholders::_1, std::placeholders::_2);
            auto del_cb = std::bind(&ServiceManager::onServiceOffline, _mm_channels.get(), std::placeholders::_1, std::placeholders::_2);
            _service_discovery = std::make_shared<Discovery>(reg_host, base_service_name, put_cb, del_cb);
        }
        void make_mysql_object(const std::string &user, const std::string &pwd, 
                const std::string &host, const std::string &db,const std::string &cset,
                int port, int conn_pool_count)
        {
            _mysql_client = ODBFactory::create(user, pwd, host, db, cset, port, conn_pool_count);
        }
        void make_mq_object(const std::string &user, const std::string &pwd, const std::string &host
                        , const std::string &exchange_name, const std::string &queue_name, const std::string &binding_key)
        {
            _exchange_name = exchange_name;
            _binding_key = binding_key;
            _mq_client = std::make_shared<MQClient>(user, pwd, host);
            _mq_client->declareComponents(exchange_name, queue_name, binding_key);
        }
        //用于构造服务注册客户端对象  服务中心地址 服务名称 当前实例的外部访问地址
        void make_reg_object(const std::string &reg_host, const std::string server_name, const std::string &access_host)
        {
            _reg_client = std::make_shared<Registry>(reg_host);
            _reg_client->registry(server_name, access_host);
        }
        void make_rpc_server(uint16_t port, uint32_t timeout, uint8_t num_threads)
        {        
            if(!_mq_client)
            {
                LOG_ERROR("消息队列模块未初始化");
                abort();
            }   
            if(!_mm_channels)
            {
                LOG_ERROR("信道管理对象未初始化");
                abort();
            }   
            if (!_mysql_client)
            {
                LOG_ERROR("mysql!客户端对象未构造");
                abort();
            }
            _rpc_server = std::make_shared<brpc::Server>();
            TransmiteServiceImpl *transmite_service = new TransmiteServiceImpl(_user_service_name
                            , _mm_channels, _mysql_client, _exchange_name, _binding_key, _mq_client); 
            int ret = _rpc_server->AddService(transmite_service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
            if(ret == -1)
            {
                LOG_ERROR("添加rpc服务失败!");
                abort();
            }
            brpc::ServerOptions options;
            options.idle_timeout_sec = timeout;  //连接空闲超时时间-超时后连接被关闭
            options.num_threads = num_threads;        //io线程数量
            //4.启动服务器
            ret = _rpc_server->Start(port, &options);
            if(ret == -1)
            {
                LOG_ERROR("启动rpc服务器失败!");
                abort();
            }

        }
        TransmiteServer::ptr build()
        {
            //_rpc_server 构造前 其余的得构造好
            if(!_service_discovery)
            {
                LOG_ERROR("服务发现对象未初始化");
                abort();
            }
            if(!_reg_client)
            {
                LOG_ERROR("服务发现对象未初始化");
                abort();
            }
            if(!_rpc_server)
            {
                LOG_ERROR("rpc服务对象未初始化");
                abort();
            }

            TransmiteServer::ptr server = std::make_shared<TransmiteServer>(_mysql_client, _service_discovery, _reg_client, _rpc_server);
            return server;
        }
    private:
        std::string _user_service_name;
        Discovery::ptr _service_discovery; //服务发现对象
        Registry::ptr _reg_client;  //注册中心
        std::shared_ptr<odb::core::database> _mysql_client; //数据库连接
        std::shared_ptr<brpc::Server> _rpc_server;       //rpc服务
        std::string _exchange_name;
        std::string _binding_key;
        MQClient::ptr _mq_client; //消息队列客户端
        ServiceManager::ptr _mm_channels; //信道管理对象
};
}